package com.sunday.common.mq.kafka.study.spring.e6_Receiving_Messages_2_Manual_Ack;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.adapter.ConsumerRecordMetadata;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.kafka.support.KafkaHeaders;
import org.springframework.kafka.support.SendResult;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.Duration;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * {@link org.springframework.boot.autoconfigure.kafka.KafkaAutoConfiguration}
 */
@Slf4j
@RestController
@RequestMapping("/test")
public class MQApi {

    @Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;

    @Autowired
    private KafkaListenerEndpointRegistry endpointRegistry;

    @GetMapping("/send/{topic}/{key}")
    public void send(@PathVariable String topic, @PathVariable String key) throws ExecutionException, InterruptedException {
        CompletableFuture<SendResult<Object, Object>> future;
        future = kafkaTemplate.send(topic, key, key);
        log.info("{}", future.get());
    }

    @GetMapping("/batch/{topic}")
    public void batch(@PathVariable String topic) {
        CompletableFuture<SendResult<Object, Object>> future;
        for (int i = 0; i < 3; i++) {
            future = kafkaTemplate.send(topic, i + "", i + "");
            future.whenComplete((result, throwable) -> log.info("[{}] result : {}, throwable : {}", topic, result, throwable));
        }
    }

    @KafkaListener(
            /** 容器的唯一标识符 */
            id = "myId_2",
            /** 条目可以是“主题名称”，“属性占位符键”或“表达式” */
//            topics = "topic2",
            /**
             * 当提供时，覆盖消费者工厂中的客户端id属性, 配置。为每个容器实例添加后缀('-n')以确保
             * [Consumer clientId=TEST-0, groupId=myId_2]
            */
            clientIdPrefix = "TEST2",
            /** 是否自动启动 */
            autoStartup = "${listen.auto.start:true}",
            /**
             * 并发数量
             * myId_2: partitions assigned: [topic2-2]
             * myId_2: partitions assigned: [topic2-0]
             * myId_2: partitions assigned: [topic2-1]
            */
//            concurrency = "${listen.concurrency:3}",
            /**
             * https://docs.spring.io/spring-kafka/docs/current/reference/html/#manual-assignment
             * 当使用手动topic/partition时，这个监听器的topicPartitions
            */
            topicPartitions =
                    {
//                            @TopicPartition(topic = "topic1", partitions = { "0", "1" }),
//                            @TopicPartition(topic = "topic2", partitions = "0",
//                                    /** partition 分区, initialOffset 偏移值 */
//                                    partitionOffsets = @PartitionOffset(partition = "2", initialOffset = "15"))
                            // 2.5.5 开始，您可以将初始偏移量应用于所有分配的分区
//                            @TopicPartition(topic = "topic2", partitions = { "0", "1" },
//                                    partitionOffsets = @PartitionOffset(partition = "*", initialOffset = "0")
//                            ),
                            // 2.6.4 开始，您可以指定以逗号分隔的分区列表或分区范围
                            @TopicPartition(topic = "topic2", partitions = "0, 1-2")
                            // 初始偏移量时，可以使用相同的技术
//                            @TopicPartition(topic = "topic2",
//                                    partitionOffsets = @PartitionOffset(partition = "0-1", initialOffset = "0"))
                    },
            /** 还可以为侦听器提供 . 以下示例还演示如何使用不同的容器工厂 */
            containerFactory = "kafkaManualAckListenerContainerFactory"
    )
    public void listen2(
            Acknowledgment ack,
            /** 从版本 2.5 开始，如果传入记录具有密钥，则不存在;以前，标头填充了一个值。 此更改是为了使框架与不存在值标头的约定保持一致 */
            @Payload String data,
            @Header(name = KafkaHeaders.RECEIVED_KEY, required = false) Integer key,
            @Header(KafkaHeaders.RECEIVED_PARTITION) int partition,
            @Header(KafkaHeaders.OFFSET) int offset,
            @Header(KafkaHeaders.GROUP_ID) String groupId,
            @Header(KafkaHeaders.RECEIVED_TOPIC) String topic,
            @Header(KafkaHeaders.TIMESTAMP_TYPE) String type,
            @Header(KafkaHeaders.RECEIVED_TIMESTAMP) long ts,
            /** 从版本 2.5 开始，您可以在参数中接收记录元数据，而不是使用离散标头。ConsumerRecordMetadata */
            ConsumerRecordMetadata meta,
            ConsumerRecord consumerRecord
    ) {
        log.info("topic : {}, partition : {}, offset : {}, groupId : {},  key : {}, data : {}, type : {}, ts : {}", topic, partition, offset, groupId, key, data, type, ts);
        log.info("meta : {} , % : {}", meta, Integer.parseInt(data) % 3 == 0);
        log.info("consumerRecord : {}", consumerRecord);
        // 手动确认
        if (Integer.parseInt(data) % 3 == 0) {
            if (pointer.incrementAndGet() % 3 == 0) {
                log.info("pointer : {}", pointer.get());
                ack.acknowledge();
                pointer.set(1);
            } else {
                ack.nack(Duration.ofSeconds(5));
            }
        } else {
            ack.acknowledge();
        }
    }

    private final AtomicInteger pointer = new AtomicInteger(1);

    /**
     * https://docs.spring.io/spring-kafka/docs/current/reference/html/#kafkalistener-lifecycle
     * <p>
     * 当 @KafkaListener 的 autoStartup 属性设置为 false 时，表示消费者不会自动启动。这意味着在应用程序启动时，该消费者不会立即开始监听和处理 Kafka 消息。
     * 要唤醒被设置为 autoStartup = false 的 @KafkaListener 消费者，您可以使用 KafkaListenerEndpointRegistry 类的 start() 方法来手动启动它。
     * 首先，确保您的 Spring Boot 应用程序中有一个 KafkaListenerEndpointRegistry 的实例。您可以通过依赖注入或者通过 @Autowired 注解获取它。
     * 然后，在需要启动消费者的地方，调用 KafkaListenerEndpointRegistry 实例的 start() 方法，传递相应的 @KafkaListener 的 id 或者名称作为参数。这将启动指定的消费者，使其开始监听和处理 Kafka 消息。
     * 在上述示例中，您需要替换 "listenerId" 为您要启动的具体 @KafkaListener 的 id 或者名称。
     * 通过调用 start() 方法，您可以手动启动被设置为 autoStartup = false 的消费者，使其开始接收和处理 Kafka 消息。
     */
    @GetMapping("/start")
    public void start() {
//        endpointRegistry.start();
        // 启动指定的消费者
        endpointRegistry.getListenerContainer("myId_2").start();
    }

    @GetMapping("/stop")
    public void stop() {
        // 启动指定的消费者
        endpointRegistry.getListenerContainer("myId_2").stop();
    }

}
